Feature: Asset outlets on DAG level for succesfull run - closes [#ISSUE](https://github.com/apache/airflow/issues/39105) by Salfiii · Pull Request #67637 · apache/airflow

Salfiii · 2026-05-28T09:02:36Z

Implemented Dag-level outles in Task SDK based on issue #39105, and kept the API consistent with how BaseOperator handles outlets.
We need a possiblity to trigger other dags by a successfully dag run and I came across the proposal in above issue that would solve our needs without manually implementing a specific fan-in/notification task at the end of every dag and it also fits the new Asset approach of airflow.
I tried to respect the guidlines in the comments of said issue.

Changes:
In task-sdk/src/airflow/sdk/definitions/dag.py:

1.Added outlets to DAG init
◦New field: outlets: list[Any]
◦Handles both a single outlet and a colection.

2.Added operator-style outlet composition on Dag
◦DAG.gt now supports dag > outlet
◦DAG.add_outlets(...) appends validated outlets

3.Wired Dag success to outlet emission
◦In attrs_post_init, if Dag outlets are set, an outlet-emission callback is appended to on_success_callback
◦Existing user callbacks are preserved
◦has_on_success_callback is updated

4.Added the callback implementation
◦Emits asset events when a Dag run succeeds
◦Only emits concrete Asset outlets
◦Includes Dag/run source metadata
◦Creates missing asset models before retrying event registration (same pattern as task outlet handling)
In airflow-core/src/airflow/assets/manager.py:

5.Extended register_asset_change(...) so Dag-level emitters can provide source fields directly. This makes outlets/events traceable to dag-runs ( I hope thats ok=
◦Added optional args:
▪source_dag_id
▪source_task_id
▪source_run_id
▪source_map_index
◦Task-instance flow is unchanged

In task-sdk/tests/task_sdk/definitions/test_dag.py:
Added tests for:
◦outlets init from single value
◦outlets init from collection
◦dag > outlet
◦rejecting invalid outlet input
◦auto-registration of Dag success outlet callback
◦preserving existing success callback when outlets are enabled

Why
The issue asks for Dag-level outlets with the same practical effect as operator outlets, but triggered when the Dag run succeeds.

closes: #ISSUE

-->

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Codex via Jetbrains AI Assistant following the guidelines

Gen-Ai was mainly used to explain the outlets structure of baseOperator and to understand on how to integrate the same approach into the DAG because the codebase is quite complex.

Important

🛠️ Maintainer triage note for @Salfiii · by @potiuk · 2026-07-02 17:38 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed (see the Pull Request quality criteria):

These required static checks are failing and need a code fix (a rerun won't help): CI image checks / Build documentation (--spellcheck-only), CI image checks / Static checks. Run them locally with pre-commit/prek and push the fixes.
The following required checks are failing: MySQL tests: core / DB-core:MySQL:8.0:3.10:Core...Serialization, Postgres tests: core / DB-core:Postgres:14:3.10:Core...Serialization, Sqlite tests: core / DB-core:Sqlite:3.10:Core...Serialization, provider distributions tests / Compat 2.11.1:P3.10:, provider distributions tests / Compat 3.0.6:P3.10:, provider distributions tests / Compat 3.1.8:P3.10:. Please investigate and push a fix.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

… and kept the API consistent with how BaseOperator handles outlets. What changed In task-sdk/src/airflow/sdk/definitions/dag.py: 1. Added outlets to DAG init ◦ New field: outlets: list[Any] ◦ Handles both a single outlet and a colection. 2. Added operator-style outlet composition on Dag ◦ DAG.__gt__ now supports dag > outlet ◦ DAG.add_outlets(...) appends validated outlets 3. Wired Dag success to outlet emission ◦ In __attrs_post_init__, if Dag outlets are set, an outlet-emission callback is appended to on_success_callback ◦ Existing user callbacks are preserved ◦ has_on_success_callback is updated 4. Added the callback implementation ◦ Emits asset events when a Dag run succeeds ◦ Only emits concrete Asset outlets ◦ Includes Dag/run source metadata ◦ Creates missing asset models before retrying event registration (same pattern as task outlet handling) In airflow-core/src/airflow/assets/manager.py: 5. Extended register_asset_change(...) so Dag-level emitters can provide source fields directly ◦ Added optional args: ▪ source_dag_id ▪ source_task_id ▪ source_run_id ▪ source_map_index ◦ Task-instance flow is unchanged ◦ This lets Dag-level events carry proper provenance without pretending to come from a task In task-sdk/tests/task_sdk/definitions/test_dag.py: 6. Added tests for: ◦ outlets init from single value ◦ outlets init from collection ◦ dag > outlet ◦ rejecting invalid outlet input ◦ auto-registration of Dag success outlet callback ◦ preserving existing success callback when outlets are enabled Why The issue asks for Dag-level outlets with the same practical effect as operator outlets, but triggered when the Dag run succeeds. This change gives you that behavior while keeping the Dag API familiar (outlets=..., dag > outlet) and making sure emitted events are traceable back to the Dag run.

Treat string values passed through dag > outlet as a single invalid outlet instead of iterating over characters. Import Asset where Dag-level outlet event emission checks for concrete assets, and update the invalid-outlet test to exercise DAG.__gt__ without triggering ruff’s pointless-comparison rule.

boring-cyborg · 2026-05-28T09:02:42Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

potiuk · 2026-06-01T12:04:14Z

@Salfiii A few things need addressing before review — see our Pull Request quality criteria.

❌ Pre-commit / static checks. See docs.

No rush.

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

potiuk · 2026-06-09T08:27:55Z

@Salfiii Converting to draft — this PR does not yet meet our Pull Request quality criteria.

❌ Pre-commit / static checks. See docs.
❌ Unit tests. See docs.

See the linked criteria for how to fix each item, then mark the PR "Ready for review". This is not a rejection — just an invitation to bring the PR up to standard. No rush.

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

Salfiii · 2026-06-10T13:14:14Z

hi @potiuk ,
thanks for taking your time to review the PR.

I´ll try to fix everything you mentioned, my problem is that i´m currently bound to a windows environment for development, which makes it quite challenging to run all the necessary checks on my side.

I can still confirm that the introduced scheduling logic works though.

best regards

Salfiii · 2026-06-10T14:45:14Z

Hi @potiuk,

I think i got all Pre-commit / static checks to run and applied some updates.

best regards

Fix serialized Dag outlets compatibility

Fix serialized Dag outlet defaults

Keep serialized Dag schema in sync

Salfiii · 2026-07-13T13:43:49Z

Hi @potiuk ,

I fixed the serialization problems, are you willing to take another look at the PR?

Thanks a lot for your patience.

Best regards

potiuk · 2026-07-20T10:18:45Z

@Salfiii This PR has a few issues that need to be addressed before it can be reviewed — please see our Pull Request quality criteria.

Issues found:

❌ Pre-commit / static checks: Failing: CI image checks / Static checks. Run prek run --from-ref main --stage pre-commit locally to reproduce and fix.

Note: Your branch is 103 commits behind main. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.

What to do next:

Fix each issue listed above.
Make sure static checks pass locally (prek run --from-ref main --stage pre-commit).
Mark the PR as "Ready for review" when you're done.

There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

Florian Salfenmoser and others added 2 commits May 27, 2026 13:29

Salfiii requested review from Lee-W, amoghrajesh, ashb, kaxil and uranusjr as code owners May 28, 2026 09:02

boring-cyborg Bot added the area:task-sdk label May 28, 2026

Merge branch 'main' into main

d3f4fe7

potiuk marked this pull request as draft June 9, 2026 08:27

Merge branch 'apache:main' into main

ecb7c3b

Salfiii and others added 4 commits June 10, 2026 15:14

Merge branch 'main' into main

7be354a

Fix type errors in DAG.test() in Task SDK

db8bf03

Merge branch 'main' of https://github.com/Salfiii/airflow

23d174c

Merge branch 'main' into main

ba26a3a

Merge branch 'main' into main

385ac39

Salfiii marked this pull request as ready for review June 11, 2026 14:52

Salfiii added 2 commits June 12, 2026 08:13

Merge branch 'main' into main

84cd4ab

Merge branch 'main' into main

6defaf7

potiuk assigned Salfiii Jun 22, 2026

F.S and others added 3 commits July 13, 2026 10:59

Fix serialized Dag outlets compatibility

724784b

Merge branch 'main' into main

558b33e

Merge pull request #1 from Salfiii/fix-serialized-dag-outlets

1e1f26d

Fix serialized Dag outlets compatibility

Salfiii requested a review from bolkedebruin as a code owner July 13, 2026 09:09

Fix serialized Dag outlet defaults

c6ffe3b

Salfiii and others added 4 commits July 13, 2026 14:04

Merge branch 'main' into main

1955648

Merge pull request #2 from Salfiii/fix-serialized-dag-outlets

cd4503a

Fix serialized Dag outlet defaults

Keep serialized Dag schema in sync

7ad33e1

Merge pull request #3 from Salfiii/fix-serialized-dag-outlets

4fbe8c2

Keep serialized Dag schema in sync

Salfiii added 3 commits July 14, 2026 10:09

Merge branch 'main' into main

054ea7f

Merge branch 'main' into main

af0f0cb

Merge branch 'main' into main

b1ec0c0

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Feature: Asset outlets on DAG level for succesfull run - closes [#ISSUE](https://github.com/apache/airflow/issues/39105)#67637

Feature: Asset outlets on DAG level for succesfull run - closes [#ISSUE](https://github.com/apache/airflow/issues/39105)#67637
Salfiii wants to merge 22 commits into
apache:mainfrom
Salfiii:main

Salfiii commented May 28, 2026 •

edited by potiuk

Loading

Uh oh!

boring-cyborg Bot commented May 28, 2026

Uh oh!

potiuk commented Jun 1, 2026

Uh oh!

potiuk commented Jun 9, 2026

Uh oh!

Salfiii commented Jun 10, 2026

Uh oh!

Salfiii commented Jun 10, 2026

Uh oh!

Salfiii commented Jul 13, 2026

Uh oh!

potiuk commented Jul 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

Salfiii commented May 28, 2026 • edited by potiuk Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Was generative AI tooling used to co-author this PR?

Uh oh!

boring-cyborg Bot commented May 28, 2026

Uh oh!

potiuk commented Jun 1, 2026

Uh oh!

potiuk commented Jun 9, 2026

Uh oh!

Salfiii commented Jun 10, 2026

Uh oh!

Salfiii commented Jun 10, 2026

Uh oh!

Salfiii commented Jul 13, 2026

Uh oh!

potiuk commented Jul 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Salfiii commented May 28, 2026 •

edited by potiuk

Loading